package com.chukun.reactor.master;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author chukun@meituan.com
 * @version 1.0.0
 * @description 主备reactor
 * @create 2022/1/25 21:18
 **/
public class Reactor implements Runnable {

    private ConcurrentLinkedQueue<AsyncHandler> events = new ConcurrentLinkedQueue<>();

    private Selector selector;

    public Reactor() throws IOException {
        this.selector = Selector.open();
    }

    public Selector getSelector() {
        return selector;
    }

    @Override
    public void run() {
        try {
            while (Thread.interrupted()) {
                AsyncHandler asyncHandler;
                while ((asyncHandler = events.poll()) != null) {
                    asyncHandler.getSocketChannel().configureBlocking(false);
                    SelectionKey sk = asyncHandler.getSocketChannel().register(selector, SelectionKey.OP_READ);
                    sk.attach(asyncHandler);
                    asyncHandler.setSk(sk);
                }
                selector.select(); //阻塞
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectionKeys.iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    //获取attach方法传入的附加对象
                    Runnable runnable = (Runnable) key.attachment();
                    if (runnable != null) {
                        runnable.run();
                    }
                    it.remove();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 注册连接请求
     *
     * @param asyncHandler
     */
    public void register(AsyncHandler asyncHandler) {
        events.offer(asyncHandler);
        selector.wakeup();
    }
}
